package spring.cloud.tasks.executor.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.util.Assert;
import spring.cloud.tasks.common.util.CronUtils;
import spring.cloud.tasks.common.util.DateUtils;
import spring.cloud.tasks.context.SpringContextHolder;
import spring.cloud.tasks.core.task.cache.TaskCache;
import spring.cloud.tasks.dispatcher.TaskDealService;
import spring.cloud.tasks.executor.bo.ProcessorParam;
import spring.cloud.tasks.executor.processor.ProcessorFactory;
import spring.cloud.tasks.executor.service.PauseOrResumeScheduleTaskService;
import spring.cloud.tasks.executor.service.ScheduleProcessor;
import spring.cloud.tasks.executor.service.TaskHeartBeatQueueService;
import spring.cloud.tasks.executor.service.TaskManager;
import spring.cloud.tasks.metadata.api.task.bo.TaskItemDefine;
import spring.cloud.tasks.metadata.api.task.entity.TaskEntity;
import spring.cloud.tasks.metadata.api.task.entity.TaskRuntimeEntity;
import spring.cloud.tasks.metadata.api.task.enums.ScheduleServerStatusEnum;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Created by xiongpu on 2019/8/3.
 */
@Slf4j
public class TaskManagerImpl implements TaskManager {
    private static Pattern SCHEDULE_TASK_ITEM_REG = Pattern.compile("\\s*:\\s*");
    /**
     * 停止调度
     */
    protected volatile boolean isStopSchedule = false;
    private TaskHeartBeatQueueService taskHeartBeatService;
    private PauseOrResumeScheduleTaskService pauseOrResumeScheduleTaskService;
    private PauseOrResumeScheduleTask resumeScheduleTask;
    private PauseOrResumeScheduleTask pauseScheduleTask;
    private List<TaskItemDefine> taskItemDefineList = new ArrayList<TaskItemDefine>();
    private ScheduleProcessor scheduleProcessor;
    private volatile TaskEntity taskEntity;
    private volatile TaskRuntimeEntity taskRuntimeEntity;
    /**
     * 队列处理器(实际调度任务的处理bean)
     */
    private TaskDealService taskDealBean;

    public TaskManagerImpl(TaskEntity taskEntity, TaskRuntimeEntity taskRuntimeEntity) throws Exception {
        this.taskEntity = taskEntity;
        this.taskRuntimeEntity = taskRuntimeEntity;
        this.taskItemDefineList = getTaskItemDefineList(taskRuntimeEntity.getTaskItems());
        this.taskDealBean = initTaskDealBean();
    }

    public static List<TaskItemDefine> getTaskItemDefineList(String taskItemsString) {
        List<TaskItemDefine> taskItemDefineList = new ArrayList<>();
        if (StringUtils.isBlank(taskItemsString)) {
            return taskItemDefineList;
        }
        List<String> taskItemStringList = Arrays.asList(taskItemsString.split(","));
        for (int i = 0; i < taskItemStringList.size(); i++) {
            TaskItemDefine taskItemDefine = new TaskItemDefine();
            String taskItemString = taskItemStringList.get(i);
            if (taskItemString == null) {
                taskItemDefineList.add(null);
                continue;
            }
            Matcher matcher = SCHEDULE_TASK_ITEM_REG.matcher(taskItemString);
            if (matcher.find()) {
                taskItemDefine.setTaskItemId(taskItemString.substring(0, matcher.start()).trim());
                taskItemDefine.setParameter(taskItemString.substring(matcher.end(), taskItemString.length() - 1).trim());
            } else {
                taskItemDefine.setTaskItemId(taskItemString);
            }
            taskItemDefineList.add(taskItemDefine);
        }
        return taskItemDefineList;
    }

    /**
     * 重新获取taskDealBean重置变量
     *
     * @throws Exception
     */
    private TaskDealService initTaskDealBean() throws Exception {
        Object dealBean = null;
        try {
            dealBean = SpringContextHolder.getBean(this.taskEntity.getDealBeanName());
        } catch (Exception e) {
            log.error("获取调度处理Bean异常,请检查相关配置(任务Bean配置和调度系统相关组件是否有打开.)", e);
        }
        if (dealBean == null) {
            throw new Exception("SpringBean " + this.taskEntity.getDealBeanName() + " 不存在");
        }

        ((TaskDealService) dealBean).setTaskInfo(this.taskEntity);
        return (TaskDealService) dealBean;
    }

    private void refresh() {
        String cluster = this.taskEntity.getCluster();
        String taskName = this.taskEntity.getTaskName();

        this.taskEntity = TaskCache.getTaskByClusterAndName(cluster, taskName);
        Assert.notNull(this.taskEntity, "从缓存内获取任务信息失败! cluster:" + cluster + ",taskName:" + taskName);
    }

    @Override
    public synchronized boolean startProcessor() {
        if (this.isStop()) {
            return false;
        }
        this.refresh();
        if (this.scheduleProcessor == null || this.scheduleProcessor.isExecutorEnd()) {
            ProcessorParam processorParam = ProcessorParam.builder().taskManager(this).taskDealService(this.taskDealBean).taskEntity(this.taskEntity).taskItemDefineList(this.taskItemDefineList).taskRuntimeEntity(this.taskRuntimeEntity).build();
            this.scheduleProcessor = ProcessorFactory.create(this.taskEntity.getProcessorType(), processorParam);
            return true;
        } else {
            log.debug("调度运行中,跳过");
            return false;
        }
    }

    @Override
    public boolean stopProcessor() {
        if (this.scheduleProcessor != null) {
            this.taskRuntimeEntity.setState(ScheduleServerStatusEnum.pause.toString());
            this.scheduleProcessor.stopProcessor();
            return true;
        }
        return false;
    }

    @Override
    public synchronized void stop() {
        this.isStopSchedule = true;
        if (this.scheduleProcessor != null) {
            this.scheduleProcessor.stopProcessor();
        }
        this.unRegisterDaemonTask(this.pauseOrResumeScheduleTaskService);
        this.unRegisterHeartBeatTaskAndClearRuntimeInfo(this.taskHeartBeatService);
        log.info("TaskName:{} SeqNo:{} taskItemDefineList:{} 已停止!", this.taskRuntimeEntity.getTaskName(), this.taskRuntimeEntity.getSeqNo(), this.taskRuntimeEntity.getTaskItems());
    }

    private boolean isStop() {
        return this.isStopSchedule;
    }

    @Override
    public void registerHeartBeatTask(TaskHeartBeatQueueService taskHeartBeatService) {
        this.taskHeartBeatService = taskHeartBeatService;
        this.taskHeartBeatService.add(this.taskEntity.getHeartBeatRate(), this.taskRuntimeEntity);
    }

    @Override
    public void unRegisterHeartBeatTask(TaskHeartBeatQueueService taskHeartBeatService) {
        this.taskHeartBeatService = taskHeartBeatService;
        this.taskHeartBeatService.remove(this.taskEntity.getHeartBeatRate(), this.taskRuntimeEntity, false);
    }

    private void unRegisterHeartBeatTaskAndClearRuntimeInfo(TaskHeartBeatQueueService taskHeartBeatService) {
        this.taskHeartBeatService = taskHeartBeatService;
        this.taskHeartBeatService.remove(this.taskEntity.getHeartBeatRate(), this.taskRuntimeEntity, true);
    }

    @Override
    public void registerDaemonTask(PauseOrResumeScheduleTaskService taskDaemonThreadService) {
        this.pauseOrResumeScheduleTaskService = taskDaemonThreadService;

        Date now = DateUtils.now();
        Date firstStartTime;
        try {
            firstStartTime = CronUtils.getNextDateAfterNow(taskEntity.getPermitRunStartTime(), now);
        } catch (Exception e) {
            this.taskRuntimeEntity.setState(ScheduleServerStatusEnum.error.name());
            this.taskRuntimeEntity.setMessage("解析Corn表达式异常! cornExpression:[" + taskEntity.getPermitRunStartTime() + "]" + e.getMessage());
            throw e;
        }
        this.taskRuntimeEntity.setNextRunStartTime(firstStartTime);
        this.resumeScheduleTask = new PauseOrResumeScheduleTask(this, taskEntity.getPermitRunStartTime(), taskRuntimeEntity, PauseOrResumeScheduleTask.TYPE_RESUME);
        this.pauseOrResumeScheduleTaskService.init(this.resumeScheduleTask, firstStartTime);

        if (StringUtils.isNotBlank(taskEntity.getPermitRunEndTime())) {
            Date firstEndTime = CronUtils.getNextDateAfterNow(taskEntity.getPermitRunEndTime(), firstStartTime);
            Date nowEndTime = CronUtils.getNextDateAfterNow(taskEntity.getPermitRunEndTime(), now);

            if (!nowEndTime.equals(firstEndTime) && now.before(nowEndTime)) {
                firstEndTime = nowEndTime;
            }
            this.pauseScheduleTask = new PauseOrResumeScheduleTask(this, taskEntity.getPermitRunEndTime(), taskRuntimeEntity, PauseOrResumeScheduleTask.TYPE_PAUSE);
            this.pauseOrResumeScheduleTaskService.schedule(this.pauseScheduleTask, firstEndTime);
            this.taskRuntimeEntity.setNextRunEndTime(firstEndTime);
        }
    }

    @Override
    public void unRegisterDaemonTask(PauseOrResumeScheduleTaskService taskDaemonThreadService) {
        this.pauseOrResumeScheduleTaskService = taskDaemonThreadService;
        if (this.resumeScheduleTask != null) {
            this.pauseOrResumeScheduleTaskService.cancel(this.resumeScheduleTask);
        }
        if (this.pauseScheduleTask != null) {
            this.pauseOrResumeScheduleTaskService.cancel(this.pauseScheduleTask);
        }
    }
}
